package cn.doitedu.rtdw.datasync;

import cn.doitedu.rtdw.udfs.GpsToGeohashUdf;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class JOB03_视频信息维表同步HBASE {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000);
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://doitedu:8020/rtdw_job1_ckpt");
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 建表，映射mysql中的源表
        tenv.executeSql(
                " create table video_mysql(                          "+
                        "     id int primary key not enforced,              "+
                        "     video_name string,                               "+
                        "     video_type string,                               "+
                        "     video_album string,                              "+
                        "     video_author string,                             "+
                        "     video_timelong bigint                            "+
                        " ) with (                                            "+
                        "    'connector' = 'jdbc',                            "+
                        "    'url' = 'jdbc:mysql://doitedu:3306/realtimedw',  "+
                        "    'table-name' = 'cms_video',                       "+
                        "    'username' = 'root',                             "+
                        "    'password' = 'root'                              "+
                        " )                                                   "
        );

        // 建表，映射hbase中的目标表
        tenv.executeSql(
                " create table video_hbase(                            "+
                        "     id int,                                  "+
                        " 	  f row<video_name string,video_type string,video_album string,video_author string,video_timelong bigint>   "+
                        " ) with (                                             "+
                        "     'connector' = 'hbase-2.2',                       "+
                        "     'table-name' = 'doit42_video',                    "+
                        "     'zookeeper.quorum' = 'doitedu:2181'              "+
                        " )                                                    "
        );

        // 写一个insert...select...
        tenv.createTemporaryFunction("gps_geo", GpsToGeohashUdf.class);
        tenv.executeSql(
                " insert into video_hbase               "+
                        " SELECT                               "+
                        "    id,      "+
                        "    row(video_name,video_type,video_album,video_author,video_timelong)  as f   "+
                        " from video_mysql                      "
        );


    }
}
